package com.leaf.spring.y201907.day24.role;

import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.HashMap;
import java.util.Map;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

import com.rabbitmq.client.Channel;

/**
 * 消费者
 * 
 * @Title:
 * @Description:
 * @Author: leaf
 * @Since:2019年7月25日
 * @Version:1.1.0
 */
@Component
public class Comsumer {

	Map<String, Integer> ackList = new HashMap<>();
	int i = 0;

	private Logger logger = LoggerFactory.getLogger(getClass());

//	@RabbitListener(queues = "testQueue", containerFactory = "simpleRabbitListenerContainerFactory")
	public void get(String message, Channel channel) {
		logger.info("消费者：{}", message);
		System.out.println("=============comsumer");
	}

	@RabbitListener(queues = "testQueue", containerFactory = "simpleRabbitListenerContainerFactory")
	public void get3(Message message, Channel channel) throws IOException {
		logger.info("消费者3：{}", message);
		System.out.println("=============comsumer3");
		channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
	}

	@RabbitListener(queues = "testQueue", containerFactory = "simpleRabbitListenerContainerFactory")
	public void get2(Message message, Channel channel) throws IOException {
		logger.info("消费者2：{}", message);
		try {
			Thread.sleep(1000);
		} catch (InterruptedException e) {
			e.printStackTrace();
		}
		System.out.println("=============comsumer2");
		channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
	}

	// 测试此段需要注释其他消费者
//	@RabbitListener(queues = { "testQueue", "testqueue" }, containerFactory = "simpleRabbitListenerContainerFactory")
	public void get1(Message message, Channel channel) throws IOException {
		logger.info("消费者1：{}", message);
		System.out.println("=============comsumer1");

		byte[] body = message.getBody();

		Integer count = ackList.get(new String(body, StandardCharsets.UTF_8));
		if (count == null) {
			count = 0;
		}

		if (count > 2) {
			logger.info("确认消息 {}", body);
			/**
			 * 手动确认
			 * 
			 * @Param multiple 是否批量处理
			 */
			channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
		} else {

			ackList.put(new String(body, StandardCharsets.UTF_8), count + 1);

			logger.info("否认消息 {}", body);
			/**
			 * 手动否认
			 * 
			 * @Param requeue 该消息是否退回，true：返回原队列，false：该消息作废
			 */
			channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, true);
			// 拒绝消息
//			channel.basicReject(message.getMessageProperties().getDeliveryTag(), true);
		}
	}

}
